package com.erik.lin.kafka;

import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.utils.ZkUtils;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.security.JaasUtils;

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * Created by Erik on 2017/9/25.
 */
public class KafKaTopicUtil {

    private ZkUtils zkUtils = ZkUtils.apply("192.168.8.8:2181", 30000, 30000, JaasUtils.isZkSecurityEnabled());

    public void deleteTopic(String topic) {
        AdminUtils.deleteTopic(zkUtils, topic);
        zkUtils.close();
    }

    public void createTopic(String topic, int partitions, int repfs) {
        AdminUtils.createTopic(zkUtils, topic, partitions, repfs, new Properties(), RackAwareMode.Enforced$.MODULE$);
        zkUtils.close();
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.8:9092,192.168.8.8:9093,192.168.8.8:9094");
        AdminClient client = AdminClient.create(props);
//        KafkaFuture<Void> futures = client.deleteTopics(Arrays.asList("test")).all();
//        futures.get();

        NewTopic newTopic = new NewTopic("event1tg3930b58f484b559a4afbf64f19d94dsecondApi", 3, (short)3);
        CreateTopicsResult ret = client.createTopics(Arrays.asList(newTopic));
        ret.all().get();

//        KafKaTopicUtil ktu = new KafKaTopicUtil();
//        ktu.deleteTopic("event1tg3930b58f484b559a4afbf64f19d94dsecondApi");
//        ktu.createTopic("event1tg3930b58f484b559a4afbf64f19d94dsecondApi", 3, 3);
    }

}
